{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Example 1: Basic Human-in-the-Loop Text Approval\n",
    "\n",
    "## Problem Statement\n",
    "\n",
    "Create a simple LangGraph that generates text content (like social media posts or emails) and requires human approval before the content is considered \"published\" or finalized. This demonstrates the most basic human-in-the-loop pattern where a human acts as a quality gate.\n",
    "\n",
    "## Features\n",
    "\n",
    "- Generate text content using an LLM\n",
    "- Present content to human for approval/rejection\n",
    "- Handle human feedback and regenerate if needed\n",
    "- Basic error handling and validation\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Install required packages (run this cell if packages are not installed)\n",
    "# !pip install langchain langgraph langchain-openai pytest"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "📚 Basic Human-in-the-Loop Text Approval Example\n",
      "==================================================\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "from typing import TypedDict, Annotated, List, Literal\n",
    "from langchain_core.messages import HumanMessage, AIMessage, SystemMessage\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.graph import StateGraph, END\n",
    "from langgraph.graph.message import add_messages\n",
    "import json\n",
    "from datetime import datetime\n",
    "\n",
    "# Set up OpenAI API key (make sure this is set in your environment)\n",
    "# os.environ[\"OPENAI_API_KEY\"] = \"your-api-key-here\"\n",
    "\n",
    "print('📚 Basic Human-in-the-Loop Text Approval Example')\n",
    "print('=' * 50)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Define the State\n",
    "\n",
    "The state keeps track of the conversation, approval status, and any feedback from the human.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "✅ State defined with fields:\n",
      "   - messages: conversation history\n",
      "   - generated_content: the AI-generated text\n",
      "   - approved: human approval status\n",
      "   - feedback: human feedback for improvements\n",
      "   - attempt_count: number of generation attempts\n",
      "   - max_attempts: maximum allowed attempts\n"
     ]
    }
   ],
   "source": [
    "class BasicApprovalState(TypedDict):\n",
    "    messages: Annotated[List, add_messages]\n",
    "    generated_content: str\n",
    "    approved: bool\n",
    "    feedback: str\n",
    "    attempt_count: int\n",
    "    max_attempts: int\n",
    "\n",
    "\n",
    "print('✅ State defined with fields:')\n",
    "print('   - messages: conversation history')\n",
    "print('   - generated_content: the AI-generated text')\n",
    "print('   - approved: human approval status')\n",
    "print('   - feedback: human feedback for improvements')\n",
    "print('   - attempt_count: number of generation attempts')\n",
    "print('   - max_attempts: maximum allowed attempts')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Define the Graph Nodes\n",
    "\n",
    "Each node represents a step in the approval workflow.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "✅ Nodes defined: generate_content, request_human_approval, finalize_content\n"
     ]
    }
   ],
   "source": [
    "def generate_content(state: BasicApprovalState) -> BasicApprovalState:\n",
    "    \"\"\"Generate text content using LLM.\"\"\"\n",
    "    print('🤖 Generating content...')\n",
    "\n",
    "    try:\n",
    "        messages = state['messages']\n",
    "        attempt = state.get('attempt_count', 0) + 1\n",
    "        feedback = state.get('feedback', '')\n",
    "\n",
    "        # Build system message based on attempt and feedback\n",
    "        system_content = (\n",
    "            \"You are a helpful content creator. Generate engaging, appropriate content based on the user's request.\"\n",
    "        )\n",
    "\n",
    "        if attempt > 1 and feedback:\n",
    "            system_content += f'\\n\\nPrevious attempt was not approved. User feedback: {feedback}\\nPlease improve the content based on this feedback.'\n",
    "\n",
    "        system_msg = SystemMessage(content=system_content)\n",
    "        full_messages = [system_msg] + messages\n",
    "\n",
    "        llm = ChatOpenAI(model='gpt-3.5-turbo', temperature=0.7)\n",
    "        response = llm.invoke(full_messages)\n",
    "\n",
    "        content = response.content\n",
    "        print(f'📝 Generated content (attempt {attempt}): {content[:100]}...')\n",
    "\n",
    "        return {'generated_content': content, 'attempt_count': attempt, 'messages': [response]}\n",
    "\n",
    "    except Exception as e:\n",
    "        print(f'❌ Error generating content: {e}')\n",
    "        return {\n",
    "            'generated_content': f'Error generating content: {str(e)}',\n",
    "            'attempt_count': state.get('attempt_count', 0) + 1,\n",
    "        }\n",
    "\n",
    "\n",
    "def request_human_approval(state: BasicApprovalState) -> BasicApprovalState:\n",
    "    \"\"\"Request human approval for the generated content.\"\"\"\n",
    "    print('👤 Requesting human approval...')\n",
    "\n",
    "    try:\n",
    "        content = state['generated_content']\n",
    "        attempt = state.get('attempt_count', 1)\n",
    "\n",
    "        print(f'\\n{\"=\" * 60}')\n",
    "        print(f'📋 CONTENT APPROVAL REQUEST (Attempt {attempt})')\n",
    "        print(f'{\"=\" * 60}')\n",
    "        print(f'Generated Content:\\n{content}')\n",
    "        print(f'{\"=\" * 60}')\n",
    "\n",
    "        while True:\n",
    "            decision = input('\\nDo you approve this content? (yes/no/feedback): ').strip().lower()\n",
    "\n",
    "            if decision in ['yes', 'y', 'approve']:\n",
    "                print('✅ Content approved!')\n",
    "                return {'approved': True, 'feedback': ''}\n",
    "            elif decision in ['no', 'n', 'reject']:\n",
    "                feedback = input('Please provide feedback for improvement: ').strip()\n",
    "                print(f'❌ Content rejected. Feedback: {feedback}')\n",
    "                return {'approved': False, 'feedback': feedback}\n",
    "            elif decision == 'feedback':\n",
    "                feedback = input('Please provide your feedback: ').strip()\n",
    "                print(f'📝 Feedback received: {feedback}')\n",
    "                return {'approved': False, 'feedback': feedback}\n",
    "            else:\n",
    "                print(\"Please enter 'yes', 'no', or 'feedback'\")\n",
    "\n",
    "    except Exception as e:\n",
    "        print(f'❌ Error in approval process: {e}')\n",
    "        return {'approved': False, 'feedback': f'Error in approval process: {str(e)}'}\n",
    "\n",
    "\n",
    "def finalize_content(state: BasicApprovalState) -> BasicApprovalState:\n",
    "    \"\"\"Finalize the approved content.\"\"\"\n",
    "    print('📋 Finalizing approved content...')\n",
    "\n",
    "    try:\n",
    "        content = state['generated_content']\n",
    "        timestamp = datetime.now().isoformat()\n",
    "\n",
    "        final_message = AIMessage(content=f'Content approved and finalized at {timestamp}:\\n\\n{content}')\n",
    "\n",
    "        print('✅ Content has been approved and finalized!')\n",
    "        print(f'📅 Timestamp: {timestamp}')\n",
    "\n",
    "        return {'messages': [final_message]}\n",
    "\n",
    "    except Exception as e:\n",
    "        print(f'❌ Error finalizing content: {e}')\n",
    "        return {'messages': [AIMessage(content=f'Error finalizing content: {str(e)}')]}\n",
    "\n",
    "\n",
    "print('✅ Nodes defined: generate_content, request_human_approval, finalize_content')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Define Routing Logic\n",
    "\n",
    "The routing logic determines the flow through the graph based on approval status and attempt limits.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "✅ Routing logic defined\n"
     ]
    }
   ],
   "source": [
    "def should_continue(state: BasicApprovalState) -> Literal['approve', 'regenerate', 'end']:\n",
    "    \"\"\"Determine next step based on approval status and attempt count.\"\"\"\n",
    "    approved = state.get('approved', False)\n",
    "    attempt_count = state.get('attempt_count', 0)\n",
    "    max_attempts = state.get('max_attempts', 3)\n",
    "\n",
    "    if approved:\n",
    "        print('🔀 Routing to finalization')\n",
    "        return 'approve'\n",
    "    elif attempt_count >= max_attempts:\n",
    "        print(f'🔀 Max attempts ({max_attempts}) reached, ending')\n",
    "        return 'end'\n",
    "    else:\n",
    "        print(f'🔀 Routing to regeneration (attempt {attempt_count + 1})')\n",
    "        return 'regenerate'\n",
    "\n",
    "\n",
    "def handle_max_attempts(state: BasicApprovalState) -> BasicApprovalState:\n",
    "    \"\"\"Handle case when maximum attempts are reached.\"\"\"\n",
    "    print('⚠️ Maximum attempts reached without approval')\n",
    "\n",
    "    message = AIMessage(\n",
    "        content='Maximum attempts reached. Content could not be approved. Please try with different requirements.'\n",
    "    )\n",
    "\n",
    "    return {'messages': [message]}\n",
    "\n",
    "\n",
    "print('✅ Routing logic defined')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Build the Graph\n",
    "\n",
    "Construct the LangGraph with nodes and edges.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "🏗️ Building basic approval graph...\n"
     ]
    },
    {
     "ename": "ValueError",
     "evalue": "'max_attempts' is already being used as a state key",
     "output_type": "error",
     "traceback": [
      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
      "\u001b[31mValueError\u001b[39m                                Traceback (most recent call last)",
      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[6]\u001b[39m\u001b[32m, line 43\u001b[39m\n\u001b[32m     40\u001b[39m     \u001b[38;5;28;01mreturn\u001b[39;00m graph\n\u001b[32m     42\u001b[39m \u001b[38;5;66;03m# Build the graph\u001b[39;00m\n\u001b[32m---> \u001b[39m\u001b[32m43\u001b[39m basic_graph = \u001b[43mbuild_basic_approval_graph\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[6]\u001b[39m\u001b[32m, line 11\u001b[39m, in \u001b[36mbuild_basic_approval_graph\u001b[39m\u001b[34m()\u001b[39m\n\u001b[32m      9\u001b[39m workflow.add_node(\u001b[33m\"\u001b[39m\u001b[33mapproval\u001b[39m\u001b[33m\"\u001b[39m, request_human_approval)\n\u001b[32m     10\u001b[39m workflow.add_node(\u001b[33m\"\u001b[39m\u001b[33mfinalize\u001b[39m\u001b[33m\"\u001b[39m, finalize_content)\n\u001b[32m---> \u001b[39m\u001b[32m11\u001b[39m \u001b[43mworkflow\u001b[49m\u001b[43m.\u001b[49m\u001b[43madd_node\u001b[49m\u001b[43m(\u001b[49m\u001b[33;43m\"\u001b[39;49m\u001b[33;43mmax_attempts\u001b[39;49m\u001b[33;43m\"\u001b[39;49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mhandle_max_attempts\u001b[49m\u001b[43m)\u001b[49m\n\u001b[32m     13\u001b[39m \u001b[38;5;66;03m# Set entry point\u001b[39;00m\n\u001b[32m     14\u001b[39m workflow.set_entry_point(\u001b[33m\"\u001b[39m\u001b[33mgenerate\u001b[39m\u001b[33m\"\u001b[39m)\n",
      "\u001b[36mFile \u001b[39m\u001b[32m~/miniconda3/envs/py312/lib/python3.12/site-packages/langgraph/graph/state.py:360\u001b[39m, in \u001b[36mStateGraph.add_node\u001b[39m\u001b[34m(self, node, action, defer, metadata, input, retry, cache_policy, destinations)\u001b[39m\n\u001b[32m    356\u001b[39m         \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\n\u001b[32m    357\u001b[39m             \u001b[33m\"\u001b[39m\u001b[33mNode name must be provided if action is not a function\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m    358\u001b[39m         )\n\u001b[32m    359\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m node \u001b[38;5;129;01min\u001b[39;00m \u001b[38;5;28mself\u001b[39m.channels:\n\u001b[32m--> \u001b[39m\u001b[32m360\u001b[39m     \u001b[38;5;28;01mraise\u001b[39;00m \u001b[38;5;167;01mValueError\u001b[39;00m(\u001b[33mf\u001b[39m\u001b[33m\"\u001b[39m\u001b[33m'\u001b[39m\u001b[38;5;132;01m{\u001b[39;00mnode\u001b[38;5;132;01m}\u001b[39;00m\u001b[33m'\u001b[39m\u001b[33m is already being used as a state key\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m    361\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m.compiled:\n\u001b[32m    362\u001b[39m     logger.warning(\n\u001b[32m    363\u001b[39m         \u001b[33m\"\u001b[39m\u001b[33mAdding a node to a graph that has already been compiled. This will \u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m    364\u001b[39m         \u001b[33m\"\u001b[39m\u001b[33mnot be reflected in the compiled graph.\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m    365\u001b[39m     )\n",
      "\u001b[31mValueError\u001b[39m: 'max_attempts' is already being used as a state key"
     ]
    }
   ],
   "source": [
    "def build_basic_approval_graph():\n",
    "    \"\"\"Build and compile the basic approval graph.\"\"\"\n",
    "    print('🏗️ Building basic approval graph...')\n",
    "\n",
    "    workflow = StateGraph(BasicApprovalState)\n",
    "\n",
    "    # Add nodes\n",
    "    workflow.add_node('generate', generate_content)\n",
    "    workflow.add_node('approval', request_human_approval)\n",
    "    workflow.add_node('finalize', finalize_content)\n",
    "    workflow.add_node('max_attempts', handle_max_attempts)\n",
    "\n",
    "    # Set entry point\n",
    "    workflow.set_entry_point('generate')\n",
    "\n",
    "    # Add edges\n",
    "    workflow.add_edge('generate', 'approval')\n",
    "\n",
    "    # Conditional routing from approval\n",
    "    workflow.add_conditional_edges(\n",
    "        'approval', should_continue, {'approve': 'finalize', 'regenerate': 'generate', 'end': 'max_attempts'}\n",
    "    )\n",
    "\n",
    "    # End nodes\n",
    "    workflow.add_edge('finalize', END)\n",
    "    workflow.add_edge('max_attempts', END)\n",
    "\n",
    "    # Compile the graph\n",
    "    graph = workflow.compile()\n",
    "\n",
    "    print('✅ Graph compiled successfully!')\n",
    "    print('   Flow: generate → approval → [approve/regenerate/end] → finalize/max_attempts')\n",
    "\n",
    "    return graph\n",
    "\n",
    "\n",
    "# Build the graph\n",
    "basic_graph = build_basic_approval_graph()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Test the Graph\n",
    "\n",
    "Run the graph with sample inputs to demonstrate the human-in-the-loop functionality.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def run_basic_example(user_request: str, max_attempts: int = 3):\n",
    "    \"\"\"Run the basic approval example.\"\"\"\n",
    "    print(f'\\n{\"=\" * 60}')\n",
    "    print(f'🚀 Running Basic Approval Example')\n",
    "    print(f'Request: {user_request}')\n",
    "    print(f'Max attempts: {max_attempts}')\n",
    "    print(f'{\"=\" * 60}')\n",
    "\n",
    "    initial_state = {\n",
    "        'messages': [HumanMessage(content=user_request)],\n",
    "        'generated_content': '',\n",
    "        'approved': False,\n",
    "        'feedback': '',\n",
    "        'attempt_count': 0,\n",
    "        'max_attempts': max_attempts,\n",
    "    }\n",
    "\n",
    "    try:\n",
    "        final_state = None\n",
    "        for state in basic_graph.stream(initial_state):\n",
    "            final_state = state\n",
    "\n",
    "        print('\\n' + '=' * 60)\n",
    "        print('📋 FINAL RESULT')\n",
    "        print('=' * 60)\n",
    "\n",
    "        if final_state:\n",
    "            last_state = list(final_state.values())[-1]\n",
    "            if 'messages' in last_state and last_state['messages']:\n",
    "                final_message = last_state['messages'][-1]\n",
    "                print(f'Result: {final_message.content}')\n",
    "\n",
    "            # Show statistics\n",
    "            attempts = last_state.get('attempt_count', 0)\n",
    "            approved = last_state.get('approved', False)\n",
    "            print(f'\\n📊 Statistics:')\n",
    "            print(f'   - Total attempts: {attempts}')\n",
    "            print(f'   - Final status: {\"✅ Approved\" if approved else \"❌ Not approved\"}')\n",
    "\n",
    "        print('\\n✅ Example completed successfully!')\n",
    "        return final_state\n",
    "\n",
    "    except Exception as e:\n",
    "        print(f'❌ Error running example: {e}')\n",
    "        return None\n",
    "\n",
    "\n",
    "# Example usage\n",
    "print('📝 Example ready to run!')\n",
    "print('\\nTo test the graph, run:')\n",
    "print('run_basic_example(\"Write a social media post about the benefits of reading books\")')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run an example (uncomment to test)\n",
    "result = run_basic_example('Write a social media post about the benefits of reading books')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 6: Unit Tests\n",
    "\n",
    "Test the core functionality of the graph components.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pytest\n",
    "from unittest.mock import patch, MagicMock\n",
    "\n",
    "\n",
    "def test_generate_content_basic():\n",
    "    \"\"\"Test basic content generation.\"\"\"\n",
    "    state = {'messages': [HumanMessage(content='Write a hello world message')], 'attempt_count': 0, 'feedback': ''}\n",
    "\n",
    "    with patch('langchain_openai.ChatOpenAI') as mock_llm_class:\n",
    "        mock_llm = MagicMock()\n",
    "        mock_response = MagicMock()\n",
    "        mock_response.content = 'Hello, World! This is a test message.'\n",
    "        mock_llm.invoke.return_value = mock_response\n",
    "        mock_llm_class.return_value = mock_llm\n",
    "\n",
    "        result = generate_content(state)\n",
    "\n",
    "        assert 'generated_content' in result\n",
    "        assert result['generated_content'] == 'Hello, World! This is a test message.'\n",
    "        assert result['attempt_count'] == 1\n",
    "\n",
    "    print('✅ test_generate_content_basic passed')\n",
    "\n",
    "\n",
    "def test_routing_logic():\n",
    "    \"\"\"Test routing logic for different scenarios.\"\"\"\n",
    "    # Test approved scenario\n",
    "    approved_state = {'approved': True, 'attempt_count': 1, 'max_attempts': 3}\n",
    "    assert should_continue(approved_state) == 'approve'\n",
    "\n",
    "    # Test max attempts reached\n",
    "    max_attempts_state = {'approved': False, 'attempt_count': 3, 'max_attempts': 3}\n",
    "    assert should_continue(max_attempts_state) == 'end'\n",
    "\n",
    "    # Test regeneration needed\n",
    "    regenerate_state = {'approved': False, 'attempt_count': 1, 'max_attempts': 3}\n",
    "    assert should_continue(regenerate_state) == 'regenerate'\n",
    "\n",
    "    print('✅ test_routing_logic passed')\n",
    "\n",
    "\n",
    "def test_error_handling():\n",
    "    \"\"\"Test error handling in content generation.\"\"\"\n",
    "    state = {'messages': [HumanMessage(content='Test message')], 'attempt_count': 0}\n",
    "\n",
    "    with patch('langchain_openai.ChatOpenAI') as mock_llm_class:\n",
    "        mock_llm_class.side_effect = Exception('API Error')\n",
    "\n",
    "        result = generate_content(state)\n",
    "\n",
    "        assert 'Error generating content' in result['generated_content']\n",
    "        assert result['attempt_count'] == 1\n",
    "\n",
    "    print('✅ test_error_handling passed')\n",
    "\n",
    "\n",
    "# Run the tests\n",
    "print('🧪 Running unit tests...')\n",
    "test_generate_content_basic()\n",
    "test_routing_logic()\n",
    "test_error_handling()\n",
    "print('\\n✅ All unit tests passed!')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "py312",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
